package bsf.util;

import bsf.base.BsfException;
import bsf.base.CallBack;
import bsf.system.thread.ThreadPool;
import com.sun.org.apache.xalan.internal.utils.FeatureManager;

import java.lang.reflect.Array;
import java.util.*;
import java.util.concurrent.*;

/**
 * Created by chejiangyi on 2017/4/7.
 */
public class ThreadUtil {
    public static void parallelFor(final ExecutorService threadPool,int parallelCount, Object[] array,  final CallBack.Action1<Object> action) {
        //任务队列
        Queue<Object> queuetasks = new LinkedList();
        queuetasks.addAll(Arrays.asList(array));

        while (queuetasks.size() > 0) {
            //运行任务列表
            final ArrayList<Object> runningtasks = new ArrayList<>();
            Object task;
            for (int i = 0; i < parallelCount; i++) {
                if ((task = queuetasks.poll()) != null) {
                    runningtasks.add(task);
                }else
                {
                    break;
                }
            }

            final CountDownLatch latch = new CountDownLatch(runningtasks.size());
            ArrayList<Future<String>> result = new ArrayList<>();
            for (final Object obj :runningtasks) {
                Future<String> future = threadPool.submit(new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        try {
                            action.invoke(obj);
                        } catch (Exception exp) {
                            throw exp;
                        } finally {
                            latch.countDown();
                        }
                        return "";
                    }
                });
                result.add(future);
            }

            try {
                latch.await();
            } catch (InterruptedException exp) {
                exp.printStackTrace();
            }
            for(Future<String> f:result)
            {
                try {
                    f.get();
                }
                catch (Exception exp)
                {
                    throw new BsfException("parallelFor并行执行出错",exp,ThreadUtil.class);
                }
            }
        }
    }
}
